package org.ricks.net;

import org.ricks.ioc.Message;
import org.ricks.log.Logger;
import org.ricks.net.handler.MessageProcessor;
import org.ricks.net.transport.UdpAioSession;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 消息分发器
 */
public class WorkDispatcher implements Runnable {
    public final static RequestTask EXECUTE_TASK_OR_SHUTDOWN = new RequestTask(null, null);
    private final BlockingQueue<RequestTask> taskQueue = new LinkedBlockingQueue<>();
    private final MessageProcessor processor;

    public WorkDispatcher(MessageProcessor processor) {
        this.processor = processor;
    }

    @Override
    public void run() {
        while (true) {
            try {
                RequestTask unit = taskQueue.take();
                if (unit == EXECUTE_TASK_OR_SHUTDOWN) {
                    Logger.info("shutdown thread:{}", Thread.currentThread());
                    break;
                }
                processor.process(unit.session, unit.message);
                unit.session.writeBuffer().flush();
            } catch (InterruptedException e) {
                Logger.info("InterruptedException", e);
            } catch (Exception e) {
                Logger.error(e.getClass().getName(), e);
            }
        }
    }

    /**
     * 任务分发
     */
    public void dispatch(UdpAioSession session, Message request) {
        dispatch(new RequestTask(session, request));
    }

    /**
     * 任务分发
     */
    public void dispatch(RequestTask requestTask) {
        taskQueue.offer(requestTask);
    }

    static class RequestTask<K,T> {
        private Message<K,T> message;
        private AioSession session;

        public RequestTask(UdpAioSession session, Message request) {
            this.session = session;
            this.message = request;
        }
    }
}
